package cn.itcast.xc.analysis.demo

import cn.itcast.xc.analysis.common.EtlEnvironment
import cn.itcast.xc.utils.HbaseUtils
import org.elasticsearch.spark._

import scala.collection.mutable.ArrayBuffer

/**
 * <p>
 * hbase、es 查询练习
 * </p>
 **/
object Query {

  def main(args: Array[String]): Unit = {
    // es 表名
    val esTableName = "demo/doc"
    // hbase表名
    val hbaseTableName = "demo"

    val spark = EtlEnvironment.getSparkSession(this.getClass.getSimpleName, esTableName)
    val sc = spark.sparkContext

    // 进行查询
    // 查询es
    val query =
    """
      |{"query":{"match":{"name":"王五"}}}
      |""".stripMargin
    val rdd = sc.esRDD(esTableName, query)
    val rowKeys = new ArrayBuffer[String]()
    rdd.collect().map(p => {
      rowKeys += p._2.get("id").get.toString
    })

    val hbaseUtils = new HbaseUtils

    val result = hbaseUtils.getRowKeyList(hbaseTableName, rowKeys)

    println(result.mkString(","))

    // 关闭资源
    spark.close()

  }

}
